package com.cike.sparkstudy.sql.java;

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.IntegerType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import scala.Tuple2;

/**
 * Json综合案例
 * 查询成绩为80分以上的学生的基本信息与成绩信息
 */
public class JSONDataSource {
    public static void main(String[] args){
        SparkConf conf = new SparkConf()
                .setMaster("local")
                .setAppName("JSONDataSource");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        //针对学生成绩信息的json创建DataFrame
        DataFrame studentScoresDF = sqlContext.read().json("/developerCodes/test/jsonTest/students.json");

        //对DataFrame注册临时表并执行sql
        studentScoresDF.registerTempTable("students_scores");
        DataFrame goodStudentScoresDF = sqlContext.sql("select name,score from students_scores where score >= 80");

        //把DataFrame转换成RDD，执行Transformation操作
        List<String> goodStudentNames = goodStudentScoresDF.javaRDD().map(new Function<Row, String>() {
            @Override
            public String call(Row row) throws Exception {
                return row.getString(0);
            }
        }).collect();

        //针对学生基本信息的Json字符串的创建RDD，然后对JavaRDD<String>转换成DataFrame
        ArrayList<String> studentInfoJsons = new ArrayList<String>();
        studentInfoJsons.add("{\"name\":\"Leo\", \"age\":18}");
        studentInfoJsons.add("{\"name\":\"Marry\", \"age\":17}");
        studentInfoJsons.add("{\"name\":\"Jack\", \"age\":19}");

        JavaRDD<String> studentInfoJsonsRDD = sc.parallelize(studentInfoJsons);
        DataFrame studentInfosDF = sqlContext.read().json(studentInfoJsonsRDD);

        //对学生基本信息注册临时表，然后查询成绩大于80的学生的基本信息
        studentInfosDF.registerTempTable("student_infos");

        String sql = "select name,age from student_infos where name in (";
        int goodStudentNamesSize = goodStudentNames.size();
        for(int i = 0; i < goodStudentNamesSize; i++){
            sql += "'" + goodStudentNames.get(i)+"'";
            if(i < goodStudentNamesSize-1){
                sql += ",";
            }
        }
        sql += ")";

        DataFrame goodstudentInfoDF = sqlContext.sql(sql);

        //将学生成绩表和学生基本信息表,DataFrame转换成JavaPairRDD，执行join transformation操作
        //将DataFrame转换成JavaRDD，再map为JavaPairRDD,然后进行join
        JavaPairRDD<String, Tuple2<Integer, Integer>> goodStudentsRDD =
                goodStudentScoresDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Integer> call(Row row) throws Exception {
                        return new Tuple2<String, Integer>(row.getString(0),
                                Integer.valueOf(String.valueOf(row.getLong(1))));
                    }

                }).join(goodstudentInfoDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Row row) throws Exception {
                        return new Tuple2<String, Integer>(row.getString(0),
                                Integer.valueOf(String.valueOf(row.getLong(1))));
                    }
                }));

        //将封装在RDD中的信息转换为一个JavaRDD<Row>格式
        //将JavaRDD,转换为DataFrame

        JavaRDD<Row> goodstudentRowsRDD = goodStudentsRDD.map(new Function<Tuple2<String, Tuple2<Integer, Integer>>, Row>() {
            @Override
            public Row call(Tuple2<String, Tuple2<Integer, Integer>> tuple) throws Exception {
                return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
            }
        });

        //创建一份元数据，将JavaRDD<Row>转换为DataFrame
        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
        structFields.add(DataTypes.createStructField("score", DataTypes.IntegerType, true));
        structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
        StructType structType = DataTypes.createStructType(structFields);

        DataFrame goodstudentsDF = sqlContext.createDataFrame(goodstudentRowsRDD, structType);

        goodstudentsDF.write().format("json").save("/developerCodes/test/jsonTest/good_students");

    }

}
